/*
 * BaseDispatcher.java    0.0.1    09/dic/2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */
package org.mentalsmash.tazio.transparency.impl;

import java.io.Serializable;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;

import org.mentalsmash.tazio.commons.identifiers.NID;
import org.mentalsmash.tazio.transparency.commons.messages.TazioMessage;
import org.mentalsmash.tazio.transparency.interfaces.Dispatcher;
import org.mentalsmash.tazio.transparency.interfaces.InputMessageListener;
import org.mentalsmash.tazio.transparency.interfaces.MessageFactory;
import org.mentalsmash.tazio.transparency.interfaces.TransparencyMonitor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.Inject;
import com.google.inject.Singleton;

/**
 * This class is the base implementation for the {@link Dispatcher} interface.
 * 
 * @version 0.0.1 09/dic/2009
 * @author Enrico Savioli <savioli.enrico@gmail.com>
 */
@Singleton
public class BaseInputDispatcher implements Dispatcher {

    private final static Logger log = LoggerFactory
	    .getLogger(BaseInputDispatcher.class);

    private final Set<Class<? extends Serializable>> registeredTypes;
    private final HashMap<Class<? extends Serializable>, Set<InputMessageListener>> listeners;
    private final MessageFactory messageFactory;
    private final TransparencyMonitor tester;

    @Inject
    public BaseInputDispatcher(MessageFactory messageFactory, TransparencyMonitor tester) {

	this.registeredTypes = new HashSet<Class<? extends Serializable>>();
	this.listeners = new HashMap<Class<? extends Serializable>, Set<InputMessageListener>>();
	this.messageFactory = messageFactory;
	this.tester = tester;

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transparency.Dispatcher#handleMessage(org.mentalsmash
     * .tazio.transparencyTemp.commons.messages.TazioMessage)
     */
    @Override
    public TazioMessage handleMessage(TazioMessage message) {

	tester.Dispatcher_handleMessage();

	log.debug("Handling the message: " + message.toString());
//	System.out.println("Handling the message: " + message.toString());

	Serializable answer = null;

	Serializable payload = message.getPayload();
	NID nid = message.getMsgID().getNodeID();

	Class<? extends Serializable> msgClass = payload.getClass();

	Set<InputMessageListener> registeredListeners = listenersOf(msgClass);

	for (InputMessageListener listener : registeredListeners) {

	    answer = listener.handleMessage(nid, payload);

	}

	TazioMessage answerMessage = messageFactory.encapsulateMessage(answer);

	return answerMessage;
    }

    /**
     * This is a private method used to retrieve a {@link java.util.Set Set} of
     * the listeners interested in being dispatched messages of the specified
     * type. This method is called when a new message arrives and it has to be
     * dispatched, by retrieving the list of listeners it is possible to notify
     * each of them of the new message.
     * 
     * @param msgClass
     *            the class for which the list of interested listeners has to be
     *            retrieved.
     * @return a Set containing the components interested in receiving
     *         notifications at the arrival of new messages of the specified
     *         type.
     */
    public Set<InputMessageListener> listenersOf(
	    Class<? extends Serializable> msgClass) {

	synchronized (listeners) {

//	    System.err.println("Looking for listeners of msg type : "+msgClass);
	    
	    Set<Class<? extends Serializable>> superClasses = new HashSet<Class<? extends Serializable>>();

	    for (Class<? extends Serializable> msgType : registeredTypes) {

		try {
		    msgClass.asSubclass(msgType);
		    superClasses.add(msgType);
		} catch (ClassCastException e) {

		}

	    }

	    Set<InputMessageListener> allListeners = new HashSet<InputMessageListener>();

	    final Set<InputMessageListener> classListeners = listeners
		    .get(msgClass);

	    if (classListeners != null) {
		allListeners.addAll(classListeners);
	    }

	    for (Class<? extends Serializable> msgType : superClasses) {
		Collection<InputMessageListener> list = this.listeners
			.get(msgType);
		if (list != null) {
		    allListeners.addAll(list);
		} else {
		    throw new RuntimeException("PORCODDIOOOO!");
		}
	    }

	    return allListeners;
	}

    }

    /**
     * This is the method used to register to the dispatching system. Before
     * adding the given listener to the list it verifies whether it is already
     * contained or not, doing so every listener is added at most once to the
     * map for every arrived-message event it wants to notified of. This will
     * prevent the problem of a message dispatched more than once if a listener
     * erroneously registers twice or more times.
     */
    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transparency.Dispatcher#register(java.io.Serializable
     * , org.mentalsmash.tazio.transparency.InputMessageListener)
     */
    @Override
    public void register(Class<? extends Serializable> msgClass,
	    InputMessageListener listener) {

	log.debug("Adding " + listener.getClass() + " to the listeners of "
		+ msgClass);

	synchronized (listeners) {

	    Set<InputMessageListener> registeredListeners = listeners
		    .get(msgClass);

	    if (registeredListeners != null) {

		if (!registeredListeners.contains(listener)) {
		    registeredListeners.add(listener);
		    log.debug("Listener added");
		}

	    } else {

		registeredListeners = new HashSet<InputMessageListener>();
		registeredListeners.add(listener);

		this.listeners.put(msgClass, registeredListeners);
		this.registeredTypes.add(msgClass);

	    }

	}

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transparency.Dispatcher#unregister(java.io.Serializable
     * , org.mentalsmash.tazio.transparency.InputMessageListener)
     */
    @Override
    public void unregister(Class<? extends Serializable> msgClass,
	    InputMessageListener listener) {

	synchronized (this.listeners) {
	    if (this.listeners.containsKey(msgClass)) {
		this.listeners.get(msgClass).remove(listener);

		if (this.listeners.get(msgClass).size() == 0) {
		    this.listeners.remove(msgClass);
		    this.registeredTypes.remove(msgClass);

		}
		log.debug("Removed " + listener + " from the listeners of "
			+ msgClass);
	    }

	}

    }

}
